package cn.doitedu.lookup;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class HbaseJoin {
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5*1000);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        tableEnv.executeSql(
                "CREATE TABLE b ( " +
                        " username STRING,                              " +
                        " f ROW<phone STRING>,                          " +
                        " PRIMARY KEY (username) NOT ENFORCED           " +
                        ") WITH (                                       " +
                        " 'connector' = 'hbase-2.2',                    " +
                        " 'table-name' = 'users',                       " +
                        " 'zookeeper.quorum' = 'doitedu:2181'           " +
                        ")");

        tableEnv.executeSql(
                " CREATE TABLE a (                                           "
                        + "     username     string,                            "
                        + "     proc_time   AS PROCTIME()                       "
                        + " ) WITH (                                            "
                        + "  'connector' = 'kafka',                             "
                        + "  'topic' = 'users',                                 "
                        + "  'properties.bootstrap.servers' = 'doitedu:9092',   "
                        + "  'properties.group.id' = 'goo1',                    "
                        + "  'scan.startup.mode' = 'latest-offset',             "
                        + "  'value.format'='json',                             "
                        + "  'value.json.fail-on-missing-field'='false',        "
                        + "  'value.fields-include' = 'EXCEPT_KEY'              "
                        + " )");

        tableEnv.executeSql("SELECT " +
                        "/*+ LOOKUP('table'='b', 'retry-predicate'='lookup_miss', 'retry-strategy'='fixed_delay', 'fixed-delay'='5s','max-attempts'='3') */" +
                        "a.*,b.* FROM a LEFT JOIN b FOR SYSTEM_TIME AS OF a.proc_time AS b  ON a.username = b.username")
                .print();
    }
}
